package server

import (
	"context"
	"github.com/pingcap-incubator/tinykv/kv/storage"
	"github.com/pingcap-incubator/tinykv/proto/pkg/kvrpcpb"
)

// The functions below are Server's Raw API. (implements TinyKvServer).
// Some helper methods can be found in sever.go in the current directory

// RawGet return the corresponding Get response based on RawGetRequest's CF and Key fields
func (server *Server) RawGet(_ context.Context, req *kvrpcpb.RawGetRequest) (*kvrpcpb.RawGetResponse, error) {
	// Your Code Here (1).
	resp := new(kvrpcpb.RawGetResponse)
	reader, err := server.storage.Reader(req.Context)
	defer reader.Close()
	if err != nil {
		resp.Error = err.Error()
		return nil, err
	}

	v, err := reader.GetCF(req.Cf, req.Key)
	if err != nil {
		resp.Error = err.Error()
		return resp, err
	}

	resp.Value = v
	resp.NotFound = v == nil
	return resp, nil
}

// RawPut puts the target data into storage and returns the corresponding response
func (server *Server) RawPut(_ context.Context, req *kvrpcpb.RawPutRequest) (*kvrpcpb.RawPutResponse, error) {
	// Your Code Here (1).
	// Hint: Consider using Storage.Modify to store data to be modified
	resp := new(kvrpcpb.RawPutResponse)
	batch := []storage.Modify{{
		Data: storage.Put{Key: req.Key, Value: req.Value, Cf: req.Cf},
	}}
	err := server.storage.Write(req.Context, batch)
	if err != nil {
		resp.Error = err.Error()
	}
	return resp, err
}

// RawDelete delete the target data from storage and returns the corresponding response
func (server *Server) RawDelete(_ context.Context, req *kvrpcpb.RawDeleteRequest) (*kvrpcpb.RawDeleteResponse, error) {
	// Your Code Here (1).
	// Hint: Consider using Storage.Modify to store data to be deleted
	resp := new(kvrpcpb.RawDeleteResponse)
	batch := []storage.Modify{{
		Data: storage.Delete{Key: req.Key, Cf: req.Cf},
	}}
	err := server.storage.Write(req.Context, batch)
	if err != nil {
		resp.Error = err.Error()
	}
	return resp, err
}

// RawScan scan the data starting from the start key up to limit. and return the corresponding result
func (server *Server) RawScan(_ context.Context, req *kvrpcpb.RawScanRequest) (*kvrpcpb.RawScanResponse, error) {
	// Your Code Here (1).
	// Hint: Consider using reader.IterCF
	resp := new(kvrpcpb.RawScanResponse)
	reader, err := server.storage.Reader(req.Context)
	defer reader.Close()
	if err != nil {
		resp.Error = err.Error()
		return nil, err
	}

	it := reader.IterCF(req.Cf)
	defer it.Close()
	limit := req.Limit

	var v []byte
	for it.Seek(req.StartKey); it.Valid() && limit > 0; it.Next() {
		k := it.Item().Key()
		v, err = it.Item().Value()
		if err != nil {
			resp.Error = err.Error()
			return resp, err
		}
		resp.Kvs = append(resp.Kvs, &kvrpcpb.KvPair{Key: k, Value: v})
		limit--
	}
	return resp, nil
}
